package net.mingsoft.report_data.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import net.mingsoft.report_data.biz.ZkDataMessageBiz;
import net.mingsoft.report_data.kafka.avro.ZkReport;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Optional;

/**
 * @author gengfa
 * @date 22/03/25
 */
@ConditionalOnProperty(value = "kfk-message.listen-enabled")
@Component
@Slf4j
public class ZkReportConsumer {

    @Resource
    ZkDataMessageBiz messageBiz;

    @KafkaListener(topics = {"zk_report"})
    public void receive(ConsumerRecord<String, ZkReport> record) {
        log.info("[zk_report] receive message: {}, offset: {}", record.value(), record.offset());
        // 先将消息入库，稍后进行解析
        Optional<ZkReport> value = Optional.ofNullable(record.value());
        if (value.isPresent()) {
            ZkReport message = value.get();
            log.info("received message {}: {}", message.getMessageId(), message.toString());
            boolean b = messageBiz.saveMessage(message);
            if (b) {
                log.info("[zk_report] message {} saved", message.getMessageId());
            } else {
                log.error("[zk_report] save message {} failed", message.getMessageId());
            }
        }
    }
}
